// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: donggu

#include <atomic>
#include <cstdlib>

#include <boost/asio.hpp>
#include <boost/asio/spawn.hpp>
#include <boost/system/error_code.hpp>

#include "coin2/base/log.h"
#include "coin2/base/rate_limiter.h"

std::atomic<int> cnt;
boost::asio::io_context ioc;
boost::asio::io_context::strand strand(ioc);
std::shared_ptr<AsyncLeakyBucket> limiter;

void Coro(boost::asio::yield_context yc) {
  boost::asio::high_resolution_timer t(ioc);
  unsigned int seedp = 0;
  while (true) {
    boost::system::error_code ec;
    int ms = rand_r(&seedp) % 1800 + 200;

    t.expires_after(std::chrono::milliseconds(ms));
    t.async_wait(yc[ec]);
    if (ec) {
      throw std::runtime_error("Delay failed");
    }

    limiter->PostCoro(yc[ec]);
    if (ec) {
      throw std::runtime_error("RateLimiter failed");
    }

    cnt++;
  }
}

int main() {
  auto t1 = std::thread([&] {
    // this prevents ioc.run() to immediately return
    boost::asio::executor_work_guard<decltype(ioc.get_executor())> work {ioc.get_executor()};
    ioc.run();
  });

  limiter.reset(new AsyncLeakyBucket(&strand, 1, 1'000'000'000));  // 1 msg / 1 sec
  boost::asio::spawn(ioc, &Coro);
  while (true) {
    std::this_thread::sleep_for(std::chrono::milliseconds(200));
    printf("\r%d", cnt.load());
    fflush(stdout);
  }

  t1.join();
}
